package com.szwistar.common.comm.jsonrpc.mqtt;

import com.szwistar.common.ERR;
import com.szwistar.common.comm.jsonrpc.JsonRpcEndpointAbstract;
import com.szwistar.common.comm.jsonrpc.JsonRpcMessage;
import com.szwistar.common.comm.jsonrpc.JsonRpcMessageCall;
import com.szwistar.common.comm.jsonrpc.JsonRpcMessageNotify;
import com.szwistar.common.comm.jsonrpc.JsonRpcMessageReturn;
import com.szwistar.common.datastruct.JsonMapper;

import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.Charset;


public class MqttJsonRpcEndpointImpl extends JsonRpcEndpointAbstract {
	// 用于通信的 MqttClient
	protected MqttClient mqttClient;

	// MQTT 消息处理器
	protected final IMqttMessageListener rpcHandler = new IMqttMessageListener() {
		@Override
		public void messageArrived(String topic, MqttMessage message) {
			try {
	            // 消息解码
				byte[] bytes = message.getPayload();
				String json = new String(bytes, 0, bytes.length, Charset.forName("utf8"));
				JsonRpcMessage msg = JsonMapper.fromJson(json, JsonRpcMessage.class);

				// 调用基类接收消息处理函数
				MqttJsonRpcEndpointImpl.this.onMessageReceived(msg);
			} catch (Exception e) {
				LOG.error("MQTT RPC 接收消息处理失败！{}", e.getMessage());
			}
		}
    };

	public MqttJsonRpcEndpointImpl() {}

	@Override
	public boolean isConnected() {
		return mqttClient != null && mqttClient.isConnected();
	}

	@Override
	public void close() {
		if(mqttClient == null) return;
		try {
			mqttClient.close();
			mqttClient = null;
		} catch(Exception e) {
			// do nothing
		}
	}

	@Override
	public void listenOn(Object linkObj, String myListenTopic) throws Exception {
		if(!(linkObj instanceof MqttClient)) {
			throw new Exception("MqttJsonRpcEndpointImpl 试图在非 MqttClient 连接上进行监听！");
		}

		// 如果已经在监听，则直接返回
		if(linkObj == mqttClient && myListenTopic.equals(myLocalTopic)) { return; }

		// 如果之前已经订阅过，则取消之前的订阅
		if(mqttClient != null && myLocalTopic != null) {
			mqttClient.unsubscribe(myLocalTopic);
		}

		mqttClient = (MqttClient)linkObj;
		myLocalTopic = myListenTopic;

        if(!mqttClient.isConnected()) {
        	LOG.error("MQTT 服务未连接！");
        	return;
        }

		// 订阅发送给我的 RPC 返回值
		mqttClient.subscribe(myLocalTopic, rpcHandler);
		LOG.info("启动 MqttJsonRpcEndpoint 成功：{}", myListenTopic);
	}

	@Override
	public void send(String target, JsonRpcMessage msg, int qos) throws Exception {
        if(mqttClient == null || !mqttClient.isConnected()) {
        	LOG.error("MQTT 服务未连接！");
        	return;
        }
    	if(msg instanceof JsonRpcMessageNotify) {
    		LOGCOMM.debug("向 MQTT 主题  '{}' 发送消息通知：{}", target, msg);
    	}

		String json = JsonMapper.toJson(msg);
        MqttMessage message = new MqttMessage(json.getBytes());
        message.setQos(qos);
		mqttClient.publish(target, message);
	}

	@Override
	public JsonRpcMessageReturn handleSubscribe(JsonRpcMessageCall request) {
		return JsonRpcMessageReturn.error(request, ERR.NOT_SUPPORT, "不支持直接针对 MQTT 端点的消息消息订阅，应该使用 MQTT 自己的消息订阅机制。");
	}
}
